package com.nulldev.util.io.util;

import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;
import java.nio.file.StandardOpenOption;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

import com.nulldev.util.VariableAPI.MathUtil;
import com.nulldev.util.VariableAPI.RandomUtil;
import com.nulldev.util.concurrency.threadIt.v4.emapi.ExecutorManager;
import com.nulldev.util.concurrency.threadIt.v4.emapi.uft.Executable;
import com.nulldev.util.io.DeallocationHelper;
import com.nulldev.util.io.IOUtils;
import com.nulldev.util.logging.Logger;
import com.nulldev.util.logging.LoggerFactory;

public class BufferManager extends Thread implements Runnable {

	/*
	 * IOUtils's fastCopy ByteBuffer manager.
	 * 
	 * Completed: - Initial implementation. - Add freeIfInUse(PTR) - Fixed some
	 * stuff - Add resizing support. - Complete resizing. - Add deallocation.
	 */

	public static enum BufferType {
		HEAP("HEAP"), /* default */
		DIRECT("DMA"),
		MMAP("MMAP")

		;

		private final String str;

		private BufferType(final String str) {
			this.str = str;
		}
	}

	public static class Buffer implements Closeable {
		private static final long UNALLOCATED = Long.MIN_VALUE;
		private static final long PREALLOCATED = Long.MIN_VALUE + 1;
		private static final long MIN_PTR_VAL = 1;
		private static final long MAX_PTR_VAL = Long.MAX_VALUE;

		private final ByteBuffer buffer;
		private FileChannel fc;
		private final AtomicLong usePtr = new AtomicLong(UNALLOCATED);
		private final AtomicLong allocAt = new AtomicLong();
		private final int size;
		private final BufferType type;

		private Buffer(final int size, final boolean direct) {
			this(size, direct ? BufferType.DIRECT : BufferType.HEAP);
		}

		private Buffer(final int size, final BufferType type) {
			this.size = size;
			this.type = type;
			switch (type) {
				case DIRECT: {
					this.buffer = ByteBuffer.allocateDirect(size);
					break;
				}
				case HEAP: {
					this.buffer = ByteBuffer.allocate(size);
					break;
				}
				case MMAP: {
					try {
						this.fc = FileChannel.open(File.createTempFile("nu-io-", ".buffer").toPath(), StandardOpenOption.TRUNCATE_EXISTING,
								StandardOpenOption.WRITE, StandardOpenOption.READ);
						this.buffer = this.fc.map(MapMode.READ_WRITE, 0, size);
						break;
					} catch (IOException e) {
						throw new RuntimeException(e);
					}
				}
				default:
					throw new UnsupportedOperationException("Unknown type: " + type);
			}
		}

		public int size() {
			return this.size;
		}

		public boolean isDirect() {
			return this.type == BufferType.DIRECT;
		}

		protected Buffer preallocate() {
			if (this.usePtr.get() != UNALLOCATED && this.usePtr.get() != PREALLOCATED)
				throw new RuntimeException("Buffer is already in use!");
			this.usePtr.lazySet(PREALLOCATED);
			return this;
		}

		public Buffer use() {
			if (this.usePtr.get() != UNALLOCATED && this.usePtr.get() != PREALLOCATED)
				throw new RuntimeException("Buffer is already in use! [ptr: " + this.usePtr.get() + "]");
			this.usePtr.lazySet(RandomUtil.randLong(MIN_PTR_VAL, MAX_PTR_VAL));
			this.allocAt.lazySet(System.currentTimeMillis());
			return this;
		}

		public Buffer then(final BufferUser user) {
			if (this.usePtr.get() == UNALLOCATED || this.usePtr.get() == PREALLOCATED)
				throw new RuntimeException("Buffer has NOT been allocated! [ptr: " + this.usePtr.get() + "]");
			user.buffer(this.buffer);
			return this;
		}

		public Buffer tryPreallocate() {
			if (this.usePtr.get() != UNALLOCATED)
				return this;
			this.usePtr.lazySet(PREALLOCATED);
			this.allocAt.lazySet(System.currentTimeMillis());
			return this;
		}

		public void free() {
			if (this.usePtr.get() == UNALLOCATED || this.usePtr.get() == PREALLOCATED) {
				log.dwarn("free: Rejected due to state: " + this.usePtr.get());
				return;
			}
			this.usePtr.lazySet(UNALLOCATED);
			this.allocAt.lazySet(0);
			this.buffer.clear();
		}

		@Override
		public String toString() {
			return "Buffer[size=" + this.size + ",ptr="
					+ (this.usePtr.get() == UNALLOCATED ? "UNALLOCATED" : (this.usePtr.get() == PREALLOCATED ? "PREALLOCATED" : this.usePtr)) + ",hash="
					+ MathUtil.toHex(this.hashCode()) + ",type=" + this.type.str + "]";
		}

		public BufferType getType() {
			return this.type;
		}

		public long leasedAt() {
			return this.allocAt.get();
		}

		@Override
		public void close() throws IOException {
			IOUtils.closeQuietly(this.fc);
		}

		public boolean inUse() {
			return this.usePtr.get() != UNALLOCATED;
		}

		private void __forceUnallocate() {
			this.usePtr.lazySet(UNALLOCATED);
			this.allocAt.lazySet(0);
		}

		public boolean isPreallocated() {
			return this.usePtr.get() == PREALLOCATED;
		}
	}

	public static interface BufferUser {
		public void buffer(final ByteBuffer buf);
	}

	public static interface BufferCallback {
		public void buffer(final Buffer buf);
	}

	private static final Logger log = LoggerFactory.getLoggerDT(false);
	private boolean allowResize;

	private DeallocationHelper dec;

	private final int defaultSize;
	private final int defaultAmount;
	private final BufferType bufferType;
	private final boolean useDirect, useMMAP;
	private final CircularQueue<Buffer> BUFFERS;

	public BufferManager(final int size, final int amount) {
		super("NU3-BUF");
		if (size <= 0)
			throw new IllegalArgumentException("size must be greater than 0!");
		if (amount <= 0)
			throw new IllegalArgumentException("amount must be greater than 0!");
		this.defaultSize = size;
		this.defaultAmount = amount;
		this.BUFFERS = new CircularQueue<Buffer>(this.defaultAmount);
		try {
			this.dec = new DeallocationHelper();
			this.allowResize = true;
		} catch (Exception e) {
			this.allowResize = false;
			this.dec = null;
		}
		final boolean allowExtendedTypes = this.supportsDeallocation();
		this.useDirect = IOUtils.canUseDMA() && allowExtendedTypes;
		this.useMMAP = IOUtils.canUseMMAP() && allowExtendedTypes;
		BufferType bt = BufferType.HEAP;
		if (this.useDirect)
			bt = BufferType.DIRECT;
		if (this.useMMAP)
			bt = BufferType.MMAP;
		this.bufferType = bt;
		for (int i = 0; i < amount; i++) {
			this.BUFFERS.enqueue(new Buffer(size, bt));
		}
		if (this.allowResize)
			super.start();
	}

	public int buffers() {
		return this.BUFFERS.size();
	}

	@Override
	public synchronized void start() {
	}

	@Override
	public void run() {
		if (this.dec == null) {
			log.dwarn("No deallocator present, disabling GC.");
			return;
		}
		log.debug("Deallocator: " + this.dec.getDeallocator());
		while (true) {
			log.debug("currentBufferCount: " + this.BUFFERS.size() + ", defaultAmount: " + this.defaultAmount);
			try {
				for (int i = (this.BUFFERS.size() - 1); i >= this.defaultAmount; i--) {
					final Buffer b = this.BUFFERS.at(i);
					final boolean shouldEvict = (System.currentTimeMillis() - b.leasedAt()) > 1500L;
					final boolean canEvict = b.tryPreallocate().isPreallocated();
					log.debug("buf-evict-check: [canEvict=" + canEvict + ", shouldEvict=" + shouldEvict + "] " + b);
					if (canEvict && shouldEvict) {
						if (this.dec.hardDeallocate(b.buffer)) {
							if (!this.BUFFERS.remove(i))
								log.warn("Failed to remove buffer from queue: " + b);
							IOUtils.closeQuietly(b);
						} else {
							if (b.type != BufferType.HEAP) {
								log.warn("Failed to deallocate buffer: " + b);
							}
							b.__forceUnallocate();
						}
					} else
						continue;
				}
			} catch (Throwable e) {
				log.error(e);
				try {
					Thread.yield();
					Thread.sleep(60000L);
				} catch (InterruptedException ex) {
					Thread.interrupted();
				}
				continue;
			}
			try {
				Thread.yield();
				Thread.sleep(15000L);
			} catch (InterruptedException e) {
				Thread.interrupted();
			}
		}
	}

	public Executable<Buffer> asyncBuffer() {
		return ExecutorManager.globalSubmitLP(() -> {
			return awaitFree();
		});
	}

	public void awaitThen(final BufferCallback bufUser) {
		bufUser.buffer(awaitFree());
	}

	@SuppressWarnings("resource")
	public Buffer awaitFree() {
		while (true) {
			try {
				Buffer buf = null;
				for (int i = 0; i < this.defaultSize; i++) {
					try {
						buf = this.BUFFERS.peek();
						if (buf == null || buf.inUse())
							continue;
						else {
							log.debug("[awaitFree] peeked candidate: " + buf);
							break;
						}
					} catch (ArrayIndexOutOfBoundsException ex) {
						continue;
					}
				}
				if ((buf == null || buf.inUse()) && this.allowResize) {
					final Buffer b = new Buffer(BufferManager.this.defaultSize, BufferManager.this.bufferType).use();
					this.BUFFERS.enqueueAndEnlarge(b);
//					log.debug("Allocated new buffer " + b + " in awaitFree(). [candidate: " + buf + "]");
					return b;
				} else {
					return buf.use();
				}
			} catch (Throwable x) {
				log.error(x);
			}
		}
	}

	public boolean supportsDeallocation() {
		return this.dec != null && this.dec.getDeallocator() != null && this.dec.hasFunctioningDeallocator();
	}

	public boolean allowsDirectAllocation() {
		return this.useDirect;
	}

	public boolean allowsMMAPAllocation() {
		return this.useMMAP;
	}

	public boolean supportsResizing() {
		return this.allowResize;
	}

	public List<Buffer> bufferList() {
		return this.BUFFERS.list();
	}
}
